package com.wetao.testrxjava;

import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.util.Log;
import android.view.View;
import android.widget.Button;

import com.wetao.testrxjava.rxjava.Person;
import com.wetao.testrxjava.rxjava.Plan;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;

import io.reactivex.Notification;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.BiConsumer;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.BooleanSupplier;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.observables.GroupedObservable;
import io.reactivex.schedulers.Schedulers;

import static java.sql.DriverManager.println;

//https://juejin.im/post/6844903617124630535#heading-126

public class TestRxJavaActivity extends AppCompatActivity  implements View.OnClickListener{

    private final String TAG = "TestRxJavaActivity";
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_test_rx_java);

        Button btn_test1 = findViewById(R.id.btn_test1);
        btn_test1.setOnClickListener(this);
        Button btn_test2 = findViewById(R.id.btn_test2);
        btn_test2.setOnClickListener(this);
        Button btn_test3 = findViewById(R.id.btn_test3);
        btn_test3.setOnClickListener(this);
        Button btn_test4 = findViewById(R.id.btn_test4);
        btn_test4.setOnClickListener(this);
        Button btn_test5 = findViewById(R.id.btn_test5);
        btn_test5.setOnClickListener(this);
        Button btn_test6 = findViewById(R.id.btn_test6);
        btn_test6.setOnClickListener(this);
        Button btn_test7 = findViewById(R.id.btn_test7);
        btn_test7.setOnClickListener(this);
        Button btn_test8 = findViewById(R.id.btn_test8);
        btn_test8.setOnClickListener(this);
        Button btn_test9 = findViewById(R.id.btn_test9);
        btn_test9.setOnClickListener(this);
        Button btn_test10 = findViewById(R.id.btn_test10);
        btn_test10.setOnClickListener(this);


        Button btn_test11 = findViewById(R.id.btn_test11);
        btn_test11.setOnClickListener(this);
        Button btn_test12 = findViewById(R.id.btn_test12);
        btn_test12.setOnClickListener(this);
        Button btn_test13 = findViewById(R.id.btn_test13);
        btn_test13.setOnClickListener(this);
        Button btn_test14 = findViewById(R.id.btn_test14);
        btn_test14.setOnClickListener(this);
        Button btn_test15 = findViewById(R.id.btn_test15);
        btn_test15.setOnClickListener(this);
        Button btn_test16 = findViewById(R.id.btn_test16);
        btn_test16.setOnClickListener(this);
        Button btn_test17 = findViewById(R.id.btn_test17);
        btn_test17.setOnClickListener(this);
        Button btn_test18 = findViewById(R.id.btn_test18);
        btn_test18.setOnClickListener(this);
        Button btn_test19 = findViewById(R.id.btn_test19);
        btn_test19.setOnClickListener(this);
        Button btn_test20 = findViewById(R.id.btn_test20);
        btn_test20.setOnClickListener(this);



        Button btn_test21 = findViewById(R.id.btn_test21);
        btn_test21.setOnClickListener(this);
        Button btn_test22 = findViewById(R.id.btn_test22);
        btn_test22.setOnClickListener(this);
        Button btn_test23 = findViewById(R.id.btn_test23);
        btn_test23.setOnClickListener(this);
        Button btn_test24 = findViewById(R.id.btn_test24);
        btn_test24.setOnClickListener(this);
        Button btn_test25 = findViewById(R.id.btn_test25);
        btn_test25.setOnClickListener(this);
        Button btn_test26 = findViewById(R.id.btn_test26);
        btn_test26.setOnClickListener(this);
        Button btn_test27 = findViewById(R.id.btn_test27);
        btn_test27.setOnClickListener(this);
        Button btn_test28 = findViewById(R.id.btn_test28);
        btn_test28.setOnClickListener(this);
        Button btn_test29 = findViewById(R.id.btn_test29);
        btn_test29.setOnClickListener(this);
        Button btn_test30 = findViewById(R.id.btn_test30);
        btn_test30.setOnClickListener(this);

        Button btn_test31 = findViewById(R.id.btn_test31);
        btn_test31.setOnClickListener(this);
        Button btn_test32 = findViewById(R.id.btn_test32);
        btn_test32.setOnClickListener(this);
        Button btn_test33 = findViewById(R.id.btn_test33);
        btn_test33.setOnClickListener(this);
        Button btn_test34 = findViewById(R.id.btn_test34);
        btn_test34.setOnClickListener(this);
        Button btn_test35 = findViewById(R.id.btn_test35);
        btn_test35.setOnClickListener(this);
        Button btn_test36 = findViewById(R.id.btn_test36);
        btn_test36.setOnClickListener(this);
        Button btn_test37 = findViewById(R.id.btn_test37);
        btn_test37.setOnClickListener(this);
        Button btn_test38 = findViewById(R.id.btn_test38);
        btn_test38.setOnClickListener(this);
        Button btn_test39 = findViewById(R.id.btn_test39);
        btn_test39.setOnClickListener(this);
        Button btn_test40 = findViewById(R.id.btn_test40);
        btn_test40.setOnClickListener(this);

        Button btn_test41 = findViewById(R.id.btn_test41);
        btn_test41.setOnClickListener(this);
        Button btn_test42 = findViewById(R.id.btn_test42);
        btn_test42.setOnClickListener(this);
        Button btn_test43 = findViewById(R.id.btn_test43);
        btn_test43.setOnClickListener(this);
        Button btn_test44 = findViewById(R.id.btn_test44);
        btn_test44.setOnClickListener(this);
        Button btn_test45 = findViewById(R.id.btn_test45);
        btn_test45.setOnClickListener(this);
        Button btn_test46 = findViewById(R.id.btn_test46);
        btn_test46.setOnClickListener(this);
        Button btn_test47 = findViewById(R.id.btn_test47);
        btn_test47.setOnClickListener(this);
        Button btn_test48 = findViewById(R.id.btn_test48);
        btn_test48.setOnClickListener(this);
        Button btn_test49 = findViewById(R.id.btn_test49);
        btn_test49.setOnClickListener(this);
        Button btn_test50 = findViewById(R.id.btn_test50);
        btn_test50.setOnClickListener(this);

        Button btn_test51 = findViewById(R.id.btn_test51);
        btn_test51.setOnClickListener(this);
        Button btn_test52 = findViewById(R.id.btn_test52);
        btn_test52.setOnClickListener(this);
        Button btn_test53 = findViewById(R.id.btn_test53);
        btn_test53.setOnClickListener(this);
        Button btn_test54 = findViewById(R.id.btn_test54);
        btn_test54.setOnClickListener(this);
        Button btn_test55 = findViewById(R.id.btn_test55);
        btn_test55.setOnClickListener(this);
        Button btn_test56 = findViewById(R.id.btn_test56);
        btn_test56.setOnClickListener(this);
        Button btn_test57 = findViewById(R.id.btn_test57);
        btn_test57.setOnClickListener(this);
        Button btn_test58 = findViewById(R.id.btn_test58);
        btn_test58.setOnClickListener(this);
        Button btn_test59 = findViewById(R.id.btn_test59);
        btn_test59.setOnClickListener(this);
        Button btn_test60 = findViewById(R.id.btn_test60);
        btn_test60.setOnClickListener(this);

        Button btn_test61 = findViewById(R.id.btn_test61);
        btn_test61.setOnClickListener(this);
        Button btn_test62 = findViewById(R.id.btn_test62);
        btn_test62.setOnClickListener(this);
        Button btn_test63 = findViewById(R.id.btn_test63);
        btn_test63.setOnClickListener(this);
        Button btn_test64 = findViewById(R.id.btn_test64);
        btn_test64.setOnClickListener(this);
        Button btn_test65 = findViewById(R.id.btn_test65);
        btn_test65.setOnClickListener(this);
        Button btn_test66 = findViewById(R.id.btn_test66);
        btn_test66.setOnClickListener(this);
        Button btn_test67 = findViewById(R.id.btn_test67);
        btn_test67.setOnClickListener(this);
        Button btn_test68 = findViewById(R.id.btn_test68);
        btn_test68.setOnClickListener(this);
        Button btn_test69 = findViewById(R.id.btn_test69);
        btn_test69.setOnClickListener(this);
        Button btn_test70 = findViewById(R.id.btn_test70);
        btn_test70.setOnClickListener(this);

        Button btn_test71 = findViewById(R.id.btn_test71);
        btn_test71.setOnClickListener(this);
        Button btn_test72 = findViewById(R.id.btn_test72);
        btn_test72.setOnClickListener(this);
        Button btn_test73 = findViewById(R.id.btn_test73);
        btn_test73.setOnClickListener(this);
        Button btn_test74 = findViewById(R.id.btn_test74);
        btn_test74.setOnClickListener(this);


    }

    @Override
    public void onClick(View v) {
        switch (v.getId()) {
            case R.id.btn_test1:{
                test1();
            }break;
            case R.id.btn_test2:{
                test2();
            }break;
            case R.id.btn_test3:{
                test3();
            }break;
            case R.id.btn_test4:{
                test4();
            }break;
            case R.id.btn_test5:{
                test5();
            }break;
            case R.id.btn_test6:{
                test6();
            }break;
            case R.id.btn_test7:{
                test7();
            }break;
            case R.id.btn_test8:{
                test8();
            }break;
            case R.id.btn_test9:{
                test9();
            }break;
            case R.id.btn_test10:{
                test10();
            }break;
            case R.id.btn_test11:{
                test11();
            }break;
            case R.id.btn_test12:{
                test12();
            }break;
            case R.id.btn_test13:{
                test13();
            }break;
            case R.id.btn_test14:{
                test14();
            }break;
            case R.id.btn_test15:{
                test15();
            }break;
            case R.id.btn_test16:{
                test16();
            }break;
            case R.id.btn_test17:{
                test17();
            }break;
            case R.id.btn_test18:{
                test18();
            }break;
            case R.id.btn_test19:{
                test19();
            }break;
            case R.id.btn_test20:{
                test20();
            }break;
            case R.id.btn_test21:{
                test21();
            }break;
            case R.id.btn_test22:{
                test22();
            }break;
            case R.id.btn_test23:{
                test23();
            }break;
            case R.id.btn_test24:{
                test24();
            }break;
            case R.id.btn_test25:{
                test25();
            }break;
            case R.id.btn_test26:{
                test26();
            }break;
            case R.id.btn_test27:{
                test27();
            }break;
            case R.id.btn_test28:{
                test28();
            }break;
            case R.id.btn_test29:{
                test29();
            }break;
            case R.id.btn_test30:{
                test30();
            }break;
            case R.id.btn_test31:{
                test31();
            }break;
            case R.id.btn_test32:{
                test32();
            }break;
            case R.id.btn_test33:{
                test33();
            }break;
            case R.id.btn_test34:{
                test34();
            }break;
            case R.id.btn_test35:{
                test35();
            }break;
            case R.id.btn_test36:{
                test36();
            }break;
            case R.id.btn_test37:{
                test37();
            }break;
            case R.id.btn_test38:{
                test38();
            }break;
            case R.id.btn_test39:{
                test39();
            }break;
            case R.id.btn_test40:{
                test40();
            }break;
            case R.id.btn_test41:{
                test41();
            }break;
            case R.id.btn_test42:{
                test42();
            }break;
            case R.id.btn_test43:{
                test43();
            }break;
            case R.id.btn_test44:{
                test44();
            }break;
            case R.id.btn_test45:{
                test45();
            }break;
            case R.id.btn_test46:{
                test46();
            }break;
            case R.id.btn_test47:{
                test47();
            }break;
            case R.id.btn_test48:{
                test48();
            }break;
            case R.id.btn_test49:{
                test49();
            }break;
            case R.id.btn_test50:{
                test50();
            }break;
            case R.id.btn_test51:{
                test51();
            }break;
            case R.id.btn_test52:{
                test52();
            }break;
            case R.id.btn_test53:{
                test53();
            }break;
            case R.id.btn_test54:{
                test54();
            }break;
            case R.id.btn_test55:{
                test55();
            }break;
            case R.id.btn_test56:{
                test56();
            }break;
            case R.id.btn_test57:{
                test57();
            }break;
            case R.id.btn_test58:{
                test58();
            }break;
            case R.id.btn_test59:{
                test59();
            }break;
            case R.id.btn_test60:{
                test60();
            }break;
            case R.id.btn_test61:{
                test61();
            }break;
            case R.id.btn_test62:{
                test62();
            }break;
            case R.id.btn_test63:{
                test63();
            }break;
            case R.id.btn_test64:{
                test64();
            }break;
            case R.id.btn_test65:{
                test65();
            }break;
            case R.id.btn_test66:{
                test66();
            }break;
            case R.id.btn_test67:{
                test67();
            }break;
            case R.id.btn_test68:{
                test68();
            }break;
            case R.id.btn_test69:{
                test69();
            }break;
            case R.id.btn_test70:{
                test70();
            }break;
            case R.id.btn_test71:{
                test71();
            }break;
            case R.id.btn_test72:{
                test72();
            }break;
            case R.id.btn_test73:{
                test73();
            }break;
            case R.id.btn_test74:{
                test74();
            }break;


        }
    }

    private void test1() {
        //创建被观察者
        Observable observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(final ObservableEmitter<Integer> emitter) throws Exception {
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        //订阅时只运行1次，执行顺序2 -- thread
                        Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                        emitter.onNext(1);
                        emitter.onError(new Throwable("出错1"));
                        //emitter.onError(new Throwable("出错2")); //不能调用两次，否则系统崩溃
                        emitter.onNext(2);
                        emitter.onComplete();
                        emitter.onNext(3); //不会运行
                        emitter.onComplete(); //不会运行
                        //emitter.onError(new Throwable("出错")); 不能放到onComplete后面，观察者抓取不到，报错，系统崩溃
                        Log.i(TAG, "========================= subscribe -- "); //会运行，所有打印中最后打印
                    }
                }).start();
            }
        });

        //创建观察者
        Observer observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                //创建只运行1次，执行顺序1 -- main
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

            }

            @Override
            public void onNext(Integer o) {
                //运行多次 执行顺序3 -- thread
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "-->" + o.intValue());

            }

            @Override
            public void onError(Throwable e) {
                //运行1次  -- thread
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                //运行1次  -- thread
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        };
        observable.subscribe(observer);
    }

    private void test2() {
        //将test1合并，链式结构
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(final ObservableEmitter<Integer> emitter) throws Exception {
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        //订阅时只运行1次，执行顺序2 -- thread
                        Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                        emitter.onNext(1);
                        emitter.onError(new Throwable("出错1"));
                        //emitter.onError(new Throwable("出错2")); //不能调用两次，否则系统崩溃
                        emitter.onNext(2);
                        emitter.onComplete();
                        emitter.onNext(3); //不会运行
                        emitter.onComplete(); //不会运行
                        //emitter.onError(new Throwable("出错")); 不能放到onComplete后面，观察者抓取不到，报错，系统崩溃
                        Log.i(TAG, "========================= subscribe -- "); //会运行，所有打印中最后打印
                    }
                }).start();
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                //创建只运行1次，执行顺序1 -- main
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

            }

            @Override
            public void onNext(Integer o) {
                //运行多次 执行顺序3 -- thread
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "-->" + o.intValue());

            }

            @Override
            public void onError(Throwable e) {
                //运行1次  -- thread
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                //运行1次  -- thread
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test3() {
        //just 创建一个被观察者，并发送事件，发送的事件不可以超过10个以上。
        Observable.just(1, 2, 3)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        //执行顺序1 main
                        Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        //执行顺序2 多次 main
                        Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "-->" + integer.intValue());
                    }

                    @Override
                    public void onError(Throwable e) {
                        // main
                        Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());
                    }

                    @Override
                    public void onComplete() {
                        //执行顺序3 main
                        Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());
                    }
                });
    }

    private void test4(){
        //fromArray()
        // 这个方法和 just() 类似，只不过 fromArray 可以传入多于10个的变量，并且可以传入一个数组。
        Integer array[] = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11};
        Observable.fromArray(array)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        //执行顺序1
                        Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

                    }

                    @Override
                    public void onNext(Integer integer) {
                        //执行顺序2 多次
                        Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "-->" + integer.intValue());

                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

                    }

                    @Override
                    public void onComplete() {
                        //执行顺序3
                        Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

                    }
                });
    }
    private void test5(){
        //fromCallable()
        //这里的 Callable 是 java.util.concurrent 中的 Callable，Callable 和 Runnable 的用法基本一致，只是它会返回一个结果值，这个结果值就是发给观察者的。
        Observable.fromCallable(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                //main
                //执行顺序1
                Log.i(TAG, "========================= call currentThread name: " + Thread.currentThread().getName());
                return 1;
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                //main
                //执行顺序2
                Log.i(TAG, "========================= accept currentThread name: " + Thread.currentThread().getName() + "-->" + integer.intValue());

            }
        });
    }
    private void test6(){
        //fromFuture()
        //参数中的 Future 是 java.util.concurrent 中的 Future，Future 的作用是增加了 cancel() 等方法操作 Callable，它可以通过 get() 方法来获取 Callable 返回的值。
        final FutureTask<String> futureTask = new FutureTask<String>(new Callable<String>() {
            //有Callable 和Runable
            @Override
            public String call() throws Exception {
                //执行顺序2
                Log.i(TAG, "========================= call currentThread name: " + Thread.currentThread().getName());
                return "返回结果";
            }
        });
        Observable.fromFuture(futureTask)
                .doOnSubscribe(new Consumer<Disposable>() {
                    @Override
                    public void accept(Disposable disposable) throws Exception {
                        //执行顺序1
                        Log.i(TAG, "========================= doOnSubscribe  accept currentThread name: " + Thread.currentThread().getName());
                        futureTask.run();//有很多方法
                    }})
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        //执行顺序3
                        Log.i(TAG, "========================= subscribe  accept currentThread name: " + Thread.currentThread().getName() + " --> " + s);
                    }});
    }
    private void test7(){
        //fromIterable
        //直接发送一个 List 集合数据给观察者
        List<Integer> list = new ArrayList<>();
        list.add(0);
        list.add(1);
        list.add(2);
        list.add(3);
        list.add(4);
        Observable.fromIterable(list)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        //执行顺序1
                        Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

                    }

                    @Override
                    public void onNext(Integer integer) {
                        //执行顺序2
                        Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "-->" + integer.intValue());

                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

                    }

                    @Override
                    public void onComplete() {
                        //执行顺序3
                        Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

                    }
                });
    }
    Integer test8_i = 10;
    private void test8(){
        //defer()
        //这个方法的作用就是直到被观察者被订阅后才会创建被观察者。
        //因为 defer() 只有观察者订阅的时候才会创建新的被观察者，所以每订阅一次就会打印一次，并且都是打印 i 最新的值。
        Observable<Integer> observable = Observable.defer(new Callable<ObservableSource<? extends Integer>>() {
            @Override
            public ObservableSource<? extends Integer> call() throws Exception {
                // 执行顺序1 test8_i = 30
                // 执行顺序5 test8_i = 40
                Log.i(TAG, "========================= call currentThread name: " + Thread.currentThread().getName() + "-->test8_i = " + test8_i);
                return Observable.just(test8_i);
            }
        });
        test8_i = 20;
        Observer observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                // 执行顺序2
                // 执行顺序5
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

            }

            @Override
            public void onNext(Integer integer) {
                // 执行顺序3 integer = 30
                // 执行顺序7 integer = 40
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "-->" + integer.intValue());

            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                // 执行顺序4
                // 执行顺序8
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        };
        test8_i = 30;
        observable.subscribe(observer);
        test8_i = 40;
        observable.subscribe(observer);
    }
    private void test9(){
        //timer()
        //当到指定时间后就会发送一个 0L 的值给观察者。
        Observable.timer(5, TimeUnit.SECONDS)
                .subscribe(new Observer<Long>() { //Long参数是固定的
                    @Override
                    public void onSubscribe(Disposable d) {
                        //执行顺序1
                        Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

                    }

                    @Override
                    public void onNext(Long aLong) {
                        //执行顺序2  5s后执行 aLong = 0  当前线程RxComputationThreadPool-1
                        Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "-->" + aLong.longValue());

                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

                    }

                    @Override
                    public void onComplete() {
                        //执行顺序3  当前线程RxComputationThreadPool-1
                        Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

                    }
                });
    }

    Disposable test10_d;
    private void test10(){
        //interval()
        //每隔一段时间就会发送一个事件，这个事件是从0开始，不断增1的数字。
        if(test10_d != null && !test10_d.isDisposed()) {
            test10_d.dispose();
            test10_d = null; //不需要这一句也可以
            return;
        }
        Observable.interval(4, TimeUnit.SECONDS)
                .subscribe(new Observer<Long>() { //Long是固定的
                    @Override
                    public void onSubscribe(Disposable d) {
                        //执行顺序1 main
                        test10_d = d;
                        Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

                    }

                    @Override
                    public void onNext(Long aLong) {
                        //执行顺序2 RxComputationThreadPool-1， 每过指定时间aLong加1，aLong初始值为0
                        Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "-->" + aLong.longValue());

                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

                    }

                    @Override
                    public void onComplete() {
                        //永远不会运行
                        Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

                    }
                });

    }

    private void test11(){
        //intervalRange()
        //可以指定发送事件的开始值和数量，其他与 interval() 的功能一样。
        //可以看出收到5次 onNext 事件，并且是从 2 开始的。
        Observable.intervalRange(2, 5, 6, 3, TimeUnit.SECONDS)
                .subscribe(new Observer<Long>() { //Long 固定
                    @Override
                    public void onSubscribe(Disposable d) {
                        //执行顺序1 main
                        Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

                    }

                    @Override
                    public void onNext(Long aLong) {
                        //执行顺序2 多次执行，RxComputationThreadPool 6s后开始执行这个函数，aLong初始值2，每3秒执行一次+1
                        Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "-->" + aLong.longValue());

                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

                    }

                    @Override
                    public void onComplete() {
                        //执行顺序最后， RxComputationThreadPool
                        Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

                    }
                });

    }
    private void test12(){
        //range()
        //同时发送一定范围的事件序列。从2开始执行，执行5次,同时执行，有顺序，但没有延时，都在主线程
        //rangeLong()
        //作用与 range() 一样，只是数据类型为 Long
        Observable.range(2, 5)
                .subscribe(new Observer<Integer>() { //Integer固定
                    @Override
                    public void onSubscribe(Disposable d) {
                        //执行顺序1， main
                        Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

                    }

                    @Override
                    public void onNext(Integer integer) {
                        //执行顺序2，多次执行 main
                        Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "-->" + integer.longValue());

                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

                    }

                    @Override
                    public void onComplete() {
                        //执行顺序最后
                        Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

                    }
                });
    }
    private void test13(){
        //empty() ： 直接发送 onComplete() 事件
        //never()：不发送任何事件
        //error()：发送 onError() 事件
        Observable
                //.never()
                //.error(new Throwable("出错了"))
                .empty()
                .subscribe(new Observer<Object>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

                    }

                    @Override
                    public void onNext(Object o) {
                        Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName());

                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

                    }

                    @Override
                    public void onComplete() {
                        Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

                    }
                });
    }
    private void test14(){
        //map()
        //map 可以将被观察者发送的数据类型转变成其他的类型
        Observable.just(1, 2, 3)
                .map(new Function<Integer, String>() {
                    @Override
                    public String apply(Integer integer) throws Exception {
                        //执行顺序2 main 多次执行 和onNext()交替执行
                        Log.i(TAG, "========================= apply currentThread name: " + Thread.currentThread().getName() + "--> " + integer);
                        return "I am " + integer;
                    }
                })
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        //执行顺序1 main
                        Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

                    }

                    @Override
                    public void onNext(String s) {
                        //执行顺序3 main 多次执行，在apply后执行
                        Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--> " + s);

                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

                    }

                    @Override
                    public void onComplete() {
                        //最后执行 main
                        Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

                    }
                });
    }
    private void test15(){
        //flatMap()
        //这个方法可以将事件序列中的元素进行整合加工，返回一个新的被观察者
        //flatMap() 其实与 map() 类似，但是 flatMap() 返回的是一个 Observerable
        List<Person> personList = new ArrayList<>();
        for(int i = 0; i < 2; i++) {
            List<Plan> planList = new ArrayList<>();
            for(int j = 0; j < 2; j++) {
                Plan plan = new Plan(i * j + "", "plan-" + i + "-" + j);
                List<String> actionList = new ArrayList<>();
                for(int k = 0; k < 2; k++) {
                    actionList.add("action-" + i + "-" + j + "-" + k);
                }
                plan.setActionList(actionList);
                planList.add(plan);
            }
            Person person = new Person("person" + i, planList);
            personList.add(person);
        }

        //现在有一个需求就是要将 Person 集合中的每个元素中的 Plan 的 action 打印出来。 首先用 map() 来实现这个需求看看：
        //可以看到 onNext() 用了嵌套 for 循环来实现，如果代码逻辑复杂起来的话，可能需要多重循环才可以实现。
        //test15_map(personList);

        //现在看下使用 flatMap() 实现：
        //从代码可以看出，只需要两个 flatMap() 就可以完成需求，并且代码逻辑非常清晰。
        test15_flatMap(personList);
    }
    private void test15_map(List<Person> personList) {
        Observable.fromIterable(personList)
                .map(new Function<Person, List<Plan>>() {
                    @Override
                    public List<Plan> apply(Person person) throws Exception {
                        return person.getPlanList();
                    }
                })
                .subscribe(new Observer<List<Plan>>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(List<Plan> plans) {
                        for (Plan plan: plans) {
                            List < String > planActionList = plan.getActionList();
                            for (String action: planActionList) {
                                Log.i(TAG, "==================action " + action);
                            }
                        }
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }
    private void test15_flatMap(List<Person> personList) {
        Observable.fromIterable(personList)
                .flatMap(new Function<Person, ObservableSource<Plan>>() {
                    @Override
                    public ObservableSource<Plan> apply(Person person) throws Exception {
                        //执行顺序2， 执行多次
                        Log.i(TAG, "========================= Person  apply currentThread name: " + Thread.currentThread().getName() + "--> " + person);
                        return Observable.fromIterable(person.getPlanList());
                    }
                })
                .flatMap(new Function<Plan, ObservableSource<String>>() {
                    @Override
                    public ObservableSource<String> apply(Plan plan) throws Exception {
                        //执行顺序3， 执行多次
                        Log.i(TAG, "========================= Plan  apply currentThread name: " + Thread.currentThread().getName() + "--> " + plan);
                        return Observable.fromIterable(plan.getActionList());
                    }
                })
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        //执行顺序1 都是main
                        Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

                    }

                    @Override
                    public void onNext(String s) {
                        //执行顺序4， 执行多次
                        Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--> " + s);

                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

                    }

                    @Override
                    public void onComplete() {
                        //最后执行
                        Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

                    }
                });
    }
    private void test16(){
        //concatMap()
        //concatMap() 和 flatMap() 基本上是一样的，只不过 concatMap() 转发出来的事件是有序的，而 flatMap() 是无序的。
        //为了更好的验证 flatMap 是无序的，使用了一个 delay() 方法来延迟, 否则看不出来差别
        List<Person> personList = new ArrayList<>();
        for(int i = 0; i < 2; i++) {
            List<Plan> planList = new ArrayList<>();
            for(int j = 0; j < 2; j++) {
                Plan plan = new Plan(i * j + "", "plan-" + i + "-" + j);
                List<String> actionList = new ArrayList<>();
                for(int k = 0; k < 2; k++) {
                    actionList.add("action-" + i + "-" + j + "-" + k);
                }
                plan.setActionList(actionList);
                planList.add(plan);
            }
            Person person = new Person("person" + i, planList);
            personList.add(person);
        }
        //验证 flatMap 是无序的
        //test16_flatMap(personList);
        //验证 concatMap 是有序的
        test16_concatMap(personList);
    }
    private void test16_flatMap(List<Person> personList){
        Observable.fromIterable(personList)
                .flatMap(new Function<Person, ObservableSource<Plan>>() {
                    @Override
                    public ObservableSource<Plan> apply(Person person) throws Exception {
                        if("person0".equals(person.getName())) {
                            return Observable.fromIterable(person.getPlanList()).delay(5, TimeUnit.SECONDS);
                        }
                        return Observable.fromIterable(person.getPlanList());
                    }
                })
                .subscribe(new Observer<Plan>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Plan plan) {
                        Log.i(TAG, "==================plan " + plan.getContent());
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }
    private void test16_concatMap(List<Person> personList) {
        Observable.fromIterable(personList)
                .concatMap(new Function<Person, ObservableSource<Plan>>() {
                    @Override
                    public ObservableSource<Plan> apply(Person person) throws Exception {
                        if("person0".equals(person.getName())) {
                            return Observable.fromIterable(person.getPlanList()).delay(5, TimeUnit.SECONDS);
                        }
                        return Observable.fromIterable(person.getPlanList());
                    }
                })
                .subscribe(new Observer<Plan>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Plan plan) {
                        Log.i(TAG, "==================plan " + plan.getContent());
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }
    private void test17(){
        //buffer()
        //从需要发送的事件当中获取一定数量的事件，并将这些事件放到缓冲区当中一并发出。
        //buffer 有两个参数，一个是 count，另一个 skip。count 缓冲区元素的数量，skip 就代表缓冲区满了之后，发送下一次事件序列的时候要跳过多少元素
        Observable.just(1, 2, 3, 4, 5)
                .buffer(2, 1)
                .subscribe(new Observer<List<Integer>>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

                    }

                    @Override
                    public void onNext(List<Integer> integers) {
                        Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName());
                        Log.i(TAG, "========================= 缓冲区大小: " + integers.size());
                        for(Integer i : integers) {
                            Log.i(TAG, "========================= 元素: " + i);
                        }
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

                    }

                    @Override
                    public void onComplete() {
                        Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

                    }
                });
    }
    private void test18(){
        //groupBy()
        //将发送的数据进行分组，每个分组都会返回一个被观察者
        //在 groupBy() 方法返回的参数是分组的名字，每返回一个值，那就代表会创建一个组，以上的代码就是将1~10的数据分成3组
        Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
                .groupBy(new Function<Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer) throws Exception {
                        Log.i(TAG, "========================= apply currentThread name: " + Thread.currentThread().getName() + "--> " + integer );

                        return integer%3;
                    }
                })
                .subscribe(new Observer<GroupedObservable<Integer, Integer>>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

                    }

                    @Override
                    public void onNext(GroupedObservable<Integer, Integer> integerIntegerGroupedObservable) {
                        Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName());
                        integerIntegerGroupedObservable.subscribe(new Observer<Integer>() {
                            @Override
                            public void onSubscribe(Disposable d) {
                                Log.i(TAG, "====================GroupedObservable onSubscribe ");
                            }

                            @Override
                            public void onNext(Integer integer) {
                                Log.i(TAG, "====================GroupedObservable onNext " + integer);

                            }

                            @Override
                            public void onError(Throwable e) {
                                Log.i(TAG, "====================GroupedObservable onError ");
                            }

                            @Override
                            public void onComplete() {
                                Log.i(TAG, "====================GroupedObservable onComplete ");

                            }
                        });
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

                    }

                    @Override
                    public void onComplete() {
                        Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

                    }
                });
    }
    private void test19(){
        //scan()
        //将数据以一定的逻辑聚合起来。
        Observable.just(1, 2, 3, 4, 5)
                .scan(new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer, Integer integer2) throws Exception {
                        Log.i(TAG, "====================apply integer = " + integer);
                        Log.i(TAG, "====================apply integer2 = " + integer2);
                        return integer + integer2;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.i(TAG, "====================accept " + integer);
                    }
                });

    }
    private void test20(){
        //window()
        //发送指定数量的事件时，就将这些事件分为一组。window 中的 count 的参数就是代表指定的数量，例如将 count 指定为2，那么每发2个数据就会将这2个数据分成一组。
        Observable.just(1, 2, 3, 4, 5)
                .window(2)
                .subscribe(new Observer<Observable<Integer>>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

                    }

                    @Override
                    public void onNext(Observable<Integer> integerObservable) {
                        integerObservable.subscribe(new Observer<Integer>() {
                            @Override
                            public void onSubscribe(Disposable d) {
                                Log.i(TAG, "=====================integerObservable onSubscribe ");
                            }

                            @Override
                            public void onNext(Integer integer) {
                                Log.i(TAG, "=====================integerObservable onNext " + integer);
                            }

                            @Override
                            public void onError(Throwable e) {
                                Log.i(TAG, "=====================integerObservable onError ");
                            }

                            @Override
                            public void onComplete() {
                                Log.i(TAG, "=====================integerObservable onComplete ");
                            }
                        });
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

                    }

                    @Override
                    public void onComplete() {
                        Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

                    }
                });
    }
    private void test21() {
        //concat()
        //可以将多个被观察者组合在一起，然后按照之前发送顺序发送事件。需要注意的是，concat() 最多只可以发送4个事件。
        Observable.concat(Observable.just(1, 2),
                Observable.just(3, 4),
                Observable.just(5, 6),
                Observable.just(7, 8))
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);

                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

                    }

                    @Override
                    public void onComplete() {
                        Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

                    }
                });
    }
    private void test22() {
        //concatArray()
        //与 concat() 作用一样，不过 concatArray() 可以发送多于 4 个被观察者。
        Observable.concatArray(Observable.just(1, 2),
                Observable.just(3, 4),
                Observable.just(5, 6),
                Observable.just(7, 8),
                Observable.just(9, 10))
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);

                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

                    }

                    @Override
                    public void onComplete() {
                        Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

                    }
                });
    }
    private void test23() {
        //merge()
        //这个方法月 concat() 作用基本一样，只是 concat() 是串行发送事件，而 merge() 并行发送事件
        //mergeArray() 与 merge() 的作用是一样的，只是它可以发送4个以上的被观察者，这里就不再赘述了。
        Observable
                //.concat(  //一直执行的是A
                .merge(  //AB交替执行
                        Observable.interval(1,TimeUnit.SECONDS).map(new Function<Long, String>() {
                            @Override
                            public String apply(Long integer) throws Exception {
                                Log.i(TAG, "========================= apply-A currentThread name: " + Thread.currentThread().getName() + "--" + integer);

                                return "A" + integer;
                            }
                        }),
                        Observable.interval(1, TimeUnit.SECONDS).map(new Function<Long, String>() {
                            @Override
                            public String apply(Long integer) throws Exception {
                                Log.i(TAG, "========================= apply-B currentThread name: " + Thread.currentThread().getName() + "--" + integer);

                                return "B" + integer;
                            }
                        })
                )
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

                    }

                    @Override
                    public void onNext(String s) {
                        Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + s);

                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

                    }

                    @Override
                    public void onComplete() {
                        Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

                    }
                });
    }
    private void test24() {
        //concatArrayDelayError() & mergeArrayDelayError()
        //在 concatArray() 和 mergeArray() 两个方法当中，如果其中有一个被观察者发送了一个 Error 事件，那么就会停止发送事件，如果你想 onError() 事件延迟到所有被观察者都发送完事件后再执行的话，就可以使用  concatArrayDelayError() 和 mergeArrayDelayError()
        Observable
                //.concatArray(
                .concatArrayDelayError(
                        Observable.create(new ObservableOnSubscribe<Integer>() {
                            @Override
                            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                                emitter.onNext(1);
                                emitter.onError(new NumberFormatException("出错了"));
                            }
                        }),
                        Observable.just(2, 3, 4)
                )
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);

                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

                    }

                    @Override
                    public void onComplete() {
                        Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

                    }
                });
    }
    private void test25() {
        //zip()
        //会将多个被观察者合并，根据各个被观察者发送事件的顺序一个个结合起来，最终发送的事件数量会与源 Observable 中最少事件的数量一样。
        //上面代码中有两个 Observable，第一个发送事件的数量为5个，第二个发送事件的数量为6个。现在来看下打印结果：
        //可以发现最终接收到的事件数量是5，那么为什么第二个 Observable 没有发送第6个事件呢？因为在这之前第一个 Observable 已经发送了 onComplete 事件，所以第二个 Observable 不会再发送事件。
        Observable.zip(Observable.intervalRange(1, 5, 1, 1, TimeUnit.SECONDS)
                        .map(new Function<Long, String>() {
                            @Override
                            public String apply(Long aLong) throws Exception {
                                Log.i(TAG, "========================= apply-A currentThread name: " + Thread.currentThread().getName() + "--" + aLong);

                                return "A" + aLong;
                            }
                        }),
                Observable.intervalRange(1, 6, 1, 1, TimeUnit.SECONDS)
                        .map(new Function<Long, String>() {
                            @Override
                            public String apply(Long aLong) throws Exception {
                                Log.i(TAG, "========================= apply-B currentThread name: " + Thread.currentThread().getName() + "--" + aLong);

                                return "B" + aLong;
                            }
                        }),
                new BiFunction<String, String, String>() {
                    @Override
                    public String apply(String s, String s2) throws Exception {
                        Log.i(TAG, "========================= BiFunction currentThread name: " + Thread.currentThread().getName() + "--" + s + s2);

                        return s + s2;
                    }
                }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

            }

            @Override
            public void onNext(String s) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + s);

            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test26() {
        //combineLatest() & combineLatestDelayError()
        //combineLatest() 的作用与 zip() 类似，但是 combineLatest() 发送事件的序列是与发送的时间线有关的，当 combineLatest() 中所有的 Observable 都发送了事件，只要其中有一个 Observable 发送事件，这个事件就会和其他 Observable 最近发送的事件结合起来发送，这样可能还是比较抽象，看看以下例子代码。
        //分析上面的代码，Observable A 会每隔1秒就发送一次事件，Observable B 会隔2秒发送一次事件。来看看打印结果：
        //分析上述结果可以知道，当发送 A1 事件之后，因为 B 并没有发送任何事件，所以根本不会发生结合。当 B 发送了 B1 事件之后，就会与 A 最近发送的事件 A2 结合成 A2B1，这样只有后面一有被观察者发送事件，这个事件就会与其他被观察者最近发送的事件结合起来了。
        //因为 combineLatestDelayError() 就是多了延迟发送 onError() 功能，这里就不再赘述了。
        Observable.combineLatest(
                Observable.intervalRange(1, 4, 1, 1, TimeUnit.SECONDS)
                        .map(new Function<Long, String>() {
                            @Override
                            public String apply(Long aLong) throws Exception {
                                Log.i(TAG, "========================= apply-A currentThread name: " + Thread.currentThread().getName() + "--" + aLong);

                                return "A" + aLong;
                            }
                        }),
                Observable.intervalRange(1, 5, 2, 2, TimeUnit.SECONDS)
                        .map(new Function<Long, String>() {
                            @Override
                            public String apply(Long aLong) throws Exception {
                                Log.i(TAG, "========================= apply-B currentThread name: " + Thread.currentThread().getName() + "--" + aLong);

                                return "B" + aLong;
                            }
                        }),
                new BiFunction<String, String, String>() {
                    @Override
                    public String apply(String s, String s2) throws Exception {
                        Log.i(TAG, "========================= BiFunction currentThread name: " + Thread.currentThread().getName() + "--" + s + s2);

                        return s+s2;
                    }
                }
        ).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

            }

            @Override
            public void onNext(String s) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + s);

            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test27() {
        //reduce()
        //与 scan() 操作符的作用也是将发送数据以一定逻辑聚合起来，这两个的区别在于 scan() 每处理一次数据就会将事件发送给观察者，而 reduce() 会将所有数据聚合在一起才会发送事件给观察者。
        //从结果可以看到，其实就是前2个数据聚合之后，然后再与后1个数据进行聚合，一直到没有数据为止。
        Observable.just(0, 1, 2, 3)
                .reduce(new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer, Integer integer2) throws Exception {
                        Log.i(TAG, "==================apply " + integer + "--" + integer2);
                        return integer + integer2;
                    }
                }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i(TAG, "==================accept " + integer);
            }
        });
    }
    private void test28() {
        //collect()
        //将数据收集到数据结构当中。
        Observable.just(1, 2, 3, 4)
                .collect(new Callable<ArrayList<Integer>>() {
                    @Override
                    public ArrayList<Integer> call() throws Exception {
                        Log.i(TAG, "==================collect call");
                        return new ArrayList<>();
                    }
                }, new BiConsumer<ArrayList<Integer>, Integer>() {
                    @Override
                    public void accept(ArrayList<Integer> integers, Integer integer) throws Exception {
                        Log.i(TAG, "==================BiConsumer integers " + integers);
                        Log.i(TAG, "==================BiConsumer integer " + integer);
                        integers.add(integer);
                    }
                })
                .subscribe(new Consumer<ArrayList<Integer>>() {
                    @Override
                    public void accept(ArrayList<Integer> integers) throws Exception {
                        Log.i(TAG, "==================accept " + integers);
                    }});
    }
    private void test29() {
        //startWith() & startWithArray()
        //在发送事件之前追加事件，startWith() 追加一个事件，startWithArray() 可以追加多个事件。追加的事件会先发出。
        Observable.just(5, 6, 7)
                .startWithArray(2, 3, 4)
                .startWith(1)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.i(TAG, "==================accept " + integer);
                    }
                });
        //上下这两个代码是一样的，onSubscribe->多个onNext->onComplete
        /*Observable.just(5, 6, 7)
                .startWithArray(2, 3, 4)
                .startWith(1)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.i(TAG, "==================onSubscribe ");
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "==================onNext " + "-->" + integer);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "==================onError ");
                    }

                    @Override
                    public void onComplete() {
                        Log.i(TAG, "==================onComplete ");
                    }
                });*/

    }
    private void test30() {
        //count()
        //返回被观察者发送事件的数量。
        Observable.just(2, 3, 4)
                .count()
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.i(TAG, "==================accept " + aLong);
                    }
                });
    }
    private void test31() {
        //delay()
        //延迟一段事件发送事件。
        Observable.just(1, 2, 3)
                .delay(5, TimeUnit.SECONDS)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);

                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

                    }

                    @Override
                    public void onComplete() {
                        Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

                    }
                });
    }
    private void test32() {
        //doOnEach()
        //Observable 每发送一件事件之前都会先回调这个方法。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();

            }
        }).doOnEach(new Consumer<Notification<Integer>>() {
            @Override
            public void accept(Notification<Integer> integerNotification) throws Exception {
                Log.i(TAG, "========================= accept currentThread name: " + Thread.currentThread().getName() + " -- " + integerNotification.getValue());

            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);

            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test33() {
        //doOnNext()
        //Observable 每发送 onNext() 之前都会先回调这个方法。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();

            }
        }).doOnNext(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i(TAG, "========================= accept currentThread name: " + Thread.currentThread().getName() + " -- " + integer);

            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);

            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });

    }
    private void test34() {
        //doAfterNext()
        //Observable 每发送 onNext() 之后都会回调这个方法。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();

            }
        }).doAfterNext(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i(TAG, "========================= accept currentThread name: " + Thread.currentThread().getName() + " -- " + integer);

            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);

            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test35() {
        //doOnComplete()
        //Observable 每发送 onComplete() 之前都会回调这个方法。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();

            }
        }).doOnComplete(new Action() {
            @Override
            public void run() throws Exception {
                Log.i(TAG, "========================= run currentThread name: " + Thread.currentThread().getName());

            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);

            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test36() {
        //doOnError()
        //Observable 每发送 onError() 之前都会回调这个方法。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onError(new NullPointerException("空指针异常"));

            }
        }).doOnError(new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                Log.i(TAG, "========================= doOnError currentThread name: " + Thread.currentThread().getName() + "--" + throwable.getMessage());

            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);

            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test37() {
        //doOnSubscribe()
        //Observable 每发送 onSubscribe() 之前都会回调这个方法。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();

            }
        }).doOnSubscribe(new Consumer<Disposable>() {
            @Override
            public void accept(Disposable disposable) throws Exception {
                Log.i(TAG, "========================= doOnSubscribe currentThread name: " + Thread.currentThread().getName());

            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);

            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test38() {
        //doOnDispose()
        //当调用 Disposable 的 dispose() 之后回调该方法。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();

            }
        }).doOnDispose(new Action() {
            @Override
            public void run() throws Exception {
                Log.i(TAG, "========================= doOnDispose currentThread name: " + Thread.currentThread().getName());

            }
        }).subscribe(new Observer<Integer>() {
            private Disposable d;
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
                this.d = d;
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);
                d.dispose();
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test39() {
        //doOnLifecycle()
        //在回调 onSubscribe 之前回调该方法的第一个参数的回调方法，可以使用该回调方法决定是否取消订阅。
        //doOnLifecycle() 第二个参数的回调方法的作用与 doOnDispose() 是一样的
        //可以看到当在 onNext() 方法进行取消订阅操作后，doOnDispose() 和 doOnLifecycle() 都会被回调。
        //如果使用 doOnLifecycle 进行取消订阅,可以发现 doOnDispose Action 和 doOnLifecycle Action 都没有被回调。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();

            }
        }).doOnLifecycle(new Consumer<Disposable>() {
            @Override
            public void accept(Disposable disposable) throws Exception {
                Log.i(TAG, "========================= doOnLifecycle currentThread name: " + Thread.currentThread().getName());
                //disposable.dispose(); //可以打开这句代码打印试试
            }
        }, new Action() {
            @Override
            public void run() throws Exception {
                Log.i(TAG, "========================= doOnLifecycle Action currentThread name: " + Thread.currentThread().getName());

            }
        }).doOnDispose(new Action() {
            @Override
            public void run() throws Exception {
                Log.i(TAG, "========================= doOnDispose currentThread name: " + Thread.currentThread().getName());

            }
        }).subscribe(new Observer<Integer>() {
            private Disposable d;
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
                this.d = d;
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);
                d.dispose();
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });

    }
    private void test40() {
        //doOnTerminate() & doAfterTerminate()
        //doOnTerminate 是在 onError 或者 onComplete 发送之前回调，而 doAfterTerminate 则是 onError 或者 onComplete 发送之后回调。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                //emitter.onError(new NullPointerException("空指针"));
                emitter.onComplete();

            }
        }).doOnTerminate(new Action() {
            @Override
            public void run() throws Exception {
                Log.i(TAG, "========================= doOnTerminate currentThread name: " + Thread.currentThread().getName());

            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test41() {
        //doFinally()
        //在所有事件发送完毕之后回调该方法。
        //这里可能你会有个问题，那就是 doFinally() 和 doAfterTerminate() 到底有什么区别？区别就是在于取消订阅，如果取消订阅之后 doAfterTerminate() 就不会被回调，而 doFinally() 无论怎么样都会被回调，且都会在事件序列的最后。
        //可以看到如果调用了 dispose() 方法，doAfterTerminate() 不会被回调。
        //
        //现在试试把 dispose() 注释掉看看
        //doAfterTerminate() 已经成功回调，doFinally() 还是会在事件序列的最后。

        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();

            }
        }).doFinally(new Action() {
            @Override
            public void run() throws Exception {
                Log.i(TAG, "========================= doFinally currentThread name: " + Thread.currentThread().getName());

            }
        }).doOnDispose(new Action() {
            @Override
            public void run() throws Exception {
                Log.i(TAG, "========================= doOnDispose currentThread name: " + Thread.currentThread().getName());

            }
        }).doAfterTerminate(new Action() {
            @Override
            public void run() throws Exception {
                Log.i(TAG, "========================= doAfterTerminate currentThread name: " + Thread.currentThread().getName());

            }
        }).subscribe(new Observer<Integer>() {
            private Disposable d;
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
                this.d = d;
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);
                d.dispose(); //可以注释这句看看打印
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });


    }
    private void test42() {
        ///onErrorReturn()
        //当接受到一个 onError() 事件之后回调，返回的值会回调 onNext() 方法，并正常结束该事件序列。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onError(new NullPointerException("空指针"));

            }
        }).onErrorReturn(new Function<Throwable, Integer>() {
            @Override
            public Integer apply(Throwable throwable) throws Exception {
                Log.i(TAG, "========================= onErrorReturn currentThread name: " + Thread.currentThread().getName() + "--" + throwable.getMessage());

                return 404;
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });

    }
    private void test43() {
        //onErrorResumeNext()
        //当接收到 onError() 事件时，返回一个新的 Observable，并正常结束事件序列。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onError(new NullPointerException("空指针"));

            }
        }).onErrorResumeNext(new Function<Throwable, ObservableSource<? extends Integer>>() {
            @Override
            public ObservableSource<? extends Integer> apply(Throwable throwable) throws Exception {
                Log.i(TAG, "========================= onErrorResumeNext currentThread name: " + Thread.currentThread().getName() + "--" + throwable.getMessage());
                return Observable.just(4, 5, 6);
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test44() {
        //onExceptionResumeNext()
        //与 onErrorResumeNext() 作用基本一致，但是这个方法只能捕捉 Exception。
        //从打印结果可以知道，观察者收到 onError() 事件，证明 onErrorResumeNext() 不能捕捉 Error 事件。
        //将被观察者的 e.onError(new Error("404")) 改为 e.onError(new Exception("404"))，现在看看是否能捕捉 Exception 事件：
        //从打印结果可以知道，这个方法成功捕获 Exception 事件。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                //emitter.onError(new Error("404")); //可以打开这句打印试试
                emitter.onError(new Exception("404"));

            }
        }).onExceptionResumeNext(new Observable<Integer>() {
            @Override
            protected void subscribeActual(Observer<? super Integer> observer) {
                Log.i(TAG, "========================= onExceptionResumeNext currentThread name: " + Thread.currentThread().getName());

                observer.onNext(333);
                observer.onComplete();
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test45() {
        //retry()
        //如果出现错误事件，则会重新发送所有事件序列。times 是代表重新发的次数。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                //emitter.onError(new Error("404")); //可以打开这句打印试试
                emitter.onError(new Exception("404"));

            }
        }).retry(2).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    int count = 0;
    private void test46() {
        //retryUntil()
        //出现错误事件之后，可以通过此方法判断是否继续发送事件。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                //emitter.onError(new Error("404")); //可以打开这句打印试试
                emitter.onError(new Exception("404"));

            }
        }).retryUntil(new BooleanSupplier() {
            @Override
            public boolean getAsBoolean() throws Exception {
                count++;
                Log.i(TAG, "========================= retryUntil currentThread name: " + Thread.currentThread().getName() + "--" + count);

                if(count >= 2) {
                    return true;
                }
                return false;
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test47() {
        //retryWhen()
        //当被观察者接收到异常或者错误事件时会回调该方法，这个方法会返回一个新的被观察者。如果返回的被观察者发送 Error 事件则之前的被观察者不会继续发送事件，如果发送正常事件则之前的被观察者会继续不断重试发送事件。
        //将 onError(new Exception("404")) 改为 onError(new Exception("303")) 看看打印结果
        //从结果可以看出，会不断重复发送消息。
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext("chan");
                emitter.onNext("ze");
                emitter.onNext("de");
                emitter.onError(new Exception("404"));
                //emitter.onError(new Exception("303"));
                emitter.onNext("haha");

            }
        }).retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(Observable<Throwable> throwableObservable) throws Exception {
                return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
                    @Override
                    public ObservableSource<?> apply(Throwable throwable) throws Exception {
                        Log.i(TAG, "========================= retryWhen currentThread name: " + Thread.currentThread().getName() + "--" + throwable.toString());

                        if(!throwable.toString().equals("java.lang.Exception: 404")) {
                            return Observable.just("可以忽略的异常");
                        } else {
                            return Observable.error(new Throwable("终止啦！"));
                        }
                    }
                });
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
            }

            @Override
            public void onNext(String s) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + s);
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test48() {
        //repeat()
        //重复发送被观察者的事件，times 为发送次数。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();

            }
        }).repeat(2).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test49() {
        //repeatWhen()
        //这个方法可以会返回一个新的被观察者设定一定逻辑来决定是否重复发送事件
        //这里分三种情况，如果新的被观察者返回 onComplete 或者 onError 事件，则旧的被观察者不会继续发送事件。如果被观察者返回其他事件，则会重复发送事件

        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();

            }
        }).repeat(2).repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(Observable<Object> objectObservable) throws Exception {
                return Observable.empty();
                //return Observable.error(new Exception("404"));
                //return Observable.just(4);
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });


    }
    private void test50() {
        //subscribeOn()
        //指定被观察者的线程，要注意的时，如果多次调用此方法，只有第一次有效。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                //新线程
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        }).subscribeOn(Schedulers.newThread())/*.subscribeOn(Schedulers.computation()).subscribeOn(Schedulers.newThread())多次调用只有第一次有效*/.subscribe(new Observer<Integer>() {  //Schedulers可以创建多种类型的线程
            @Override
            public void onSubscribe(Disposable d) {
                //主线程
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

            }

            @Override
            public void onNext(Integer integer) {
                //新线程
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);

            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                //新线程
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });

    }
    //Schedulers.computation( ) 用于使用计算任务，如事件循环和回调处理
    //Schedulers.immediate( ) 当前线程
    //Schedulers.io( ) 用于 IO 密集型任务，如果异步阻塞 IO 操作。
    //Schedulers.newThread( )创建一个新的线程
    //AndroidSchedulers.mainThread()Android 的 UI 线程，用于操作 UI。
    private void test51() {
        //observeOn()
        //指定观察者的线程，每指定一次就会生效一次。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());
                //新线程1，subscribeOn(Schedulers.newThread()) 这句代码切换的
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
                emitter.onComplete();
            }
        }).subscribeOn(Schedulers.newThread()).observeOn(Schedulers.newThread()).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                //新线程2，上面这句observeOn(Schedulers.newThread())代码切换的
                Log.i(TAG, "========================= flatMap currentThread name: " + Thread.currentThread().getName() + "--" + integer);

                return Observable.just("chan" + integer);
            }
        }).observeOn(AndroidSchedulers.mainThread())/*.observeOn(Schedulers.newThread())设置最后一个的线程，会切换到后面的代码*/.subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                //主线程 初始化
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());

            }

            @Override
            public void onNext(String s) {
                //主线程，上面这句observeOn(AndroidSchedulers.mainThread())代码切换的
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + s);

            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                //主线程，上面这句observeOn(AndroidSchedulers.mainThread())代码切换的
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test52() {
        //filter()
        //通过一定逻辑来过滤被观察者发送的事件，如果返回 true 则会发送事件，否则不会发送
        Observable.just(1, 2, 3).filter(new Predicate<Integer>() {
            @Override
            public boolean test(Integer integer) throws Exception {
                return integer < 3;
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test53() {
        //ofType()
        //可以过滤不符合该类型事件
        Observable.just(1, 2, 3, "chan", "zhide").ofType(Integer.class).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test54() {
        //skip()
        //跳过正序某些事件，count 代表跳过事件的数量
        //skipLast() 作用也是跳过某些事件，不过它是用来跳过正序的后面的事件，这里就不再讲解了。

        Observable.just(1, 2, 3).skip(2).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });

    }
    private void test55() {
        //distinct()
        //过滤事件序列中的重复事件。
        Observable.just(1, 2, 3, 3 ,2 ,1).distinct().subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test56() {
        //distinctUntilChanged()
        //过滤掉连续重复的事件
        Observable.just(1, 2, 3, 3 ,2 ,1).distinctUntilChanged().subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test57() {
        //take()
        //控制观察者接收的事件的数量。
        //takeLast() 的作用就是控制观察者只能接受事件序列的后面几件事情
        Observable.just(1, 2, 3, 4 ,5).take(3).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test58() {
        //debounce()
        //如果两件事件发送的时间间隔小于设定的时间间隔则前一件事件就不会发送给观察者。
        //throttleWithTimeout() 与此方法的作用一样
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.i(TAG, "========================= subscribe currentThread name: " + Thread.currentThread().getName());

                emitter.onNext(1);
                //Thread.sleep(900);
                Thread.sleep(1100);
                emitter.onNext(2);
            }
        }).debounce(1, TimeUnit.SECONDS).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
            }

            @Override
            public void onNext(Integer integer) {
                //新线程
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });
    }
    private void test59() {
        //firstElement() && lastElement()
        //firstElement() 取事件序列的第一个元素，lastElement() 取事件序列的最后一个元素。
        Observable.just(1, 2, 3, 4 ,5)/*.firstElement()*/.lastElement().subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i(TAG, "========================= accept currentThread name: " + Thread.currentThread().getName() +"--" +integer);

            }
        });
    }
    private void test60() {
        //elementAt() & elementAtOrError()
        //elementAt() 可以指定取出事件序列中事件，但是输入的 index 超出事件序列的总数的话就不会出现任何结果。这种情况下，你想发出异常信息的话就用 elementAtOrError() 。
        Observable.just(1, 2, 3, 4)/*.elementAt(0)*//*.elementAt(5)*/.elementAtOrError(1)/*.elementAtOrError(5)*/.subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.i(TAG, "========================= accept currentThread name: " + Thread.currentThread().getName() +"--" +integer);

            }
        });

    }
    private void test61() {
        //all()
        //判断事件序列是否全部满足某个事件，如果都满足则返回 true，反之则返回 false。
        Observable.just(1, 2, 3, 4)
                .all(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        //return integer < 5;
                        return integer < 4;
                    }
                })
                .subscribe(new Consumer<Boolean>() {
                    @Override
                    public void accept(Boolean aBoolean) throws Exception {
                        Log.i(TAG, "========================= accept currentThread name: " + Thread.currentThread().getName() +"--" +aBoolean);

                    }
                });
    }
    private void test62() {
        //takeWhile()
        //可以设置条件，当某个数据满足条件时就会发送该数据，反之则不发送
        Observable.just(1, 2, 3, 4)
                .takeWhile(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer < 3;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.i(TAG, "========================= accept currentThread name: " + Thread.currentThread().getName() +"--" +integer);

                    }
                });

    }
    private void test63() {
        //skipWhile()
        //可以设置条件，当某个数据满足条件时不发送该数据，反之则发送。
        Observable.just(1, 2, 3, 4)
                .skipWhile(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer < 3;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.i(TAG, "========================= accept currentThread name: " + Thread.currentThread().getName() +"--" +integer);

                    }
                });
    }
    private void test64() {
        //takeUntil()
        //可以设置条件，当事件满足此条件时，下一次的事件就不会被发送了。
        Observable.just(1, 2, 3, 4)
                .takeUntil(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer >= 2;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.i(TAG, "========================= accept currentThread name: " + Thread.currentThread().getName() +"--" +integer);

                    }
                });
    }
    private void test65() {
        //skipUntil()
        //当 skipUntil() 中的 Observable 发送事件了，原来的 Observable 才会发送事件给观察者。
        //从结果可以看出，skipUntil() 里的 Observable 并不会发送事件给观察者。
        Observable.intervalRange(1, 5, 0, 1, TimeUnit.SECONDS)
                .skipUntil(Observable.intervalRange(6, 5, 2, 1, TimeUnit.SECONDS)).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.i(TAG, "========================= onSubscribe currentThread name: " + Thread.currentThread().getName());
            }

            @Override
            public void onNext(Long l) {
                Log.i(TAG, "========================= onNext currentThread name: " + Thread.currentThread().getName() + "--" + l);
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "========================= onError currentThread name: " + Thread.currentThread().getName() + "--" + e.getMessage());

            }

            @Override
            public void onComplete() {
                Log.i(TAG, "========================= onComplete currentThread name: " + Thread.currentThread().getName());

            }
        });

    }
    private void test66() {
        //sequenceEqual()
        //判断两个 Observable 发送的事件是否相同
        Observable.sequenceEqual(Observable.just(1, 2, 3)/*, Observable.just(1, 2, 3)*/, Observable.just(1, 2, 3, 4))
                .subscribe(new Consumer<Boolean>() {
                    @Override
                    public void accept(Boolean b) throws Exception {
                        Log.i(TAG, "========================= accept currentThread name: " + Thread.currentThread().getName() +"--" +b);

                    }
                });
    }
    private void test67() {
        //contains()
        //判断事件序列中是否含有某个元素，如果有则返回 true，如果没有则返回 false。
        Observable.just(1, 2, 3, 4)
                .contains(3)
                .subscribe(new Consumer<Boolean>() {
                    @Override
                    public void accept(Boolean b) throws Exception {
                        Log.i(TAG, "========================= accept currentThread name: " + Thread.currentThread().getName() +"--" +b);

                    }
                });
    }
    private void test68() {
        //isEmpty()
        //判断事件序列是否为空。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onComplete();
            }
        })
                .isEmpty()
                .subscribe(new Consumer<Boolean>() {
                    @Override
                    public void accept(Boolean b) throws Exception {
                        Log.i(TAG, "========================= accept currentThread name: " + Thread.currentThread().getName() +"--" +b);

                    }
                });
    }
    private void test69() {
        //amb()
        //amb() 要传入一个 Observable 集合，但是只会发送最先发送事件的 Observable 中的事件，其余 Observable 将会被丢弃。
        ArrayList<Observable<Long>> list = new ArrayList<>();
        list.add(Observable.intervalRange(1, 5, 2, 1, TimeUnit.SECONDS));
        list.add(Observable.intervalRange(6, 5, 0, 1, TimeUnit.SECONDS));

        Observable.amb(list)
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.i(TAG, "========================= accept currentThread name: " + Thread.currentThread().getName() +"--" +aLong);

                    }
                });
    }
    private void test70() {
        //defaultIfEmpty()
        //如果观察者只发送一个 onComplete() 事件，则可以利用这个方法发送一个值。
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onComplete();
            }
        })
                .defaultIfEmpty(666)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.i(TAG, "========================= accept currentThread name: " + Thread.currentThread().getName() +"--" +integer);

                    }
                });

    }
    private void test71() {
        //Compose的作用：  对 Observable 进行整体的变换
        //rxjava中compose的使用
        //在安卓中我们使用rxjava进行操作的时候经常会在子线程进行一下常见的操作
        //然后会回到主线程中进行一些更新UI视图的操作
        Observable.just(1, 2).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {

                    }
                });

        //在安卓中使用rxjava进行线程切换的时候以下的两句代码我们会经常的使用到
        //.subscribeOn(Schedulers.io())
        //.observeOn(AndroidSchedulers.mainThread())
        //如果用的地方少还无所谓 但是如果用的地方多了我们就会选择用一个方法将切换线程的操作给封装起来:
    }
    private void test72() {
        //这种方法将频繁的切换线程的操作给封装起来可读性、可维护性都提高了。可是 Observable 被一个方法包起来，
        //这种方式对于 Observale 的灵活性似乎还是增添了那么点限制，那用compose来解决

        applySchedulers(Observable.just(1,2)).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {

            }
        });
    }
    public <T> Observable<T> applySchedulers(Observable<T> source) {
        return source.subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread());
    }
    private void test73() {
        //compose的用法
        Observable.just(1, 2)
                .compose(new MyTransformer())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {

                    }
                });
    }
    class MyTransformer implements ObservableTransformer{
        @Override
        public ObservableSource apply(Observable upstream) {
            return upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
        }
    }

    private void test74() {
        //首先我们知道使用rxjava执行操作的时候，基本流程大致如下，取消操作时只需要调用subscription.unsubscribe()即可
        //可以从doOnUnsubscribe()、doOnCompleted()、doOnError()这些hook入手
        Observable.just(1, 2)
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.newThread())
                .unsubscribeOn(Schedulers.newThread())
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }
	
	private void test75() {
		//Single只发射一条单一的数据，或者一条异常通知，不能发射完成通知，其中数据与通知只能发射一个。
		Single.create(new SingleOnSubscribe<Integer>() {
			@Override
			public void subscribe(SingleEmitter<Integer> e) throws Exception {
				e.onSuccess(0);
				//e.onError(new Exception("测试异常"));
			}
		}).subscribe(new SingleObserver<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onSuccess(Integer integer) {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                });
	}
	
	private void test76() {
		//Completable只发射一条完成通知，或者一条异常通知，不能发射数据，其中完成通知与异常通知只能发射一个
		//Maybe可发射一条单一的数据，以及发射一条完成通知，或者一条异常通知，其中完成通知和异常通知只能发射一个，发射数据只能在发射完成通知或者异常通知之前，否则发射数据无效。
		Completable.create(new CompletableOnSubscribe() {
			@Override
			public void subscribe(CompletableEmitter e) throws Exception {
				e.onComplete();
				//e.onError(new Exception("测试异常"));
			}
		}).subscribe(new CompletableObserver() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onComplete() {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                });
	}
	
	private void test77() {
		//Rxjava背压：被观察者发送事件的速度大于观察者接收事件的速度时，观察者内会创建一个无限制大少的缓冲池存储未接收的事件，因此当存储的事件越来越多时就会导致OOM的出现。（注：当subscribeOn与observeOn不为同一个线程时，被观察者与观察者内存在不同时长耗时任务，就会使发送与接收速度存在差异。）
		//Flowable 与 Observabley 用法相似
		//Flowable运行效率要比Observable低得多。
		//从Flowable源码查看，缓存池默认大少为：128
		/*
		   BackpressureStrategy媒体类:5种背压策略：
		       ERROR：当被观察者发送事件大于128时，观察者抛出异常并终止接收事件，但不会影响被观察者继续发送事件。
			   BUFFER：与Observable一样存在背压问题，但是接收性能比Observable低，因为BUFFER类型通过BufferAsyncEmitter添加了额外的逻辑处理，再发送至观察者。
			   DROP：每当观察者接收128事件之后，就会丢弃部分事件
			   LATEST：LATEST与DROP使用效果一样，但LATEST会保证能接收最后一个事件，而DROP则不会保证。
			   MISSING：MISSING就是没有采取背压策略的类型，效果跟Obserable一样。在设置MISSING类型时，可以配合onBackPressure相关操作符使用，也可以到达上述其他类型的处理效果。
		*/
		/*
		   onBackPressure相关操作符
		       onBackpressureBuffer ：与BUFFER类型一样效果。
			   onBackpressureDrop ：与DROP类型一样效果。
               onBackpressureLaster ：与LASTER类型一样效果。
		*/
		Flowable.create(new FlowableOnSubscribe<Integer>() {
            @Override
            public void subscribe(FlowableEmitter<Integer> e) throws Exception {
                for(int j = 0;j<=150;j++){
                    e.onNext(j);
                    Log.i(TAG," 发送数据："+j);
					int count = e.requested(); //requested打印的结果就是 剩余可接收的数量 ,它的作用就是可以检测剩余可接收的事件数量
					Log.i(TAG,count+" 发送数据："+j);
                    try{
                        Thread.sleep(50);
                    }catch (Exception ex){
                    }
                }
            }
        },BackpressureStrategy.ERROR)
		//.onBackpressureDrop()//效果与Drop类型一样
        .subscribeOn(Schedulers.newThread())
        .observeOn(Schedulers.newThread())
        .subscribe(new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                s.request(Long.MAX_VALUE); //观察者设置接收事件的数量,如果不设置接收不到事件
				//subscription = s;
                //s.request(10); //观察者设置接收事件的数量,如果不设置接收不到事件
			}
            @Override
            public void onNext(Integer integer) {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                Log.e(TAG,"onNext : "+(integer));
				//subscription.request(3); 可以动态设置观察者接收事件的数量，但不影响被观察者继续发送事件
            }
            @Override
            public void onError(Throwable t) {
                Log.e(TAG,"onError : "+t.toString());
            }
            @Override
            public void onComplete() {
                Log.e(TAG,"onComplete");
            }
        });
		
		
		
		Flowable.interval(50,TimeUnit.MILLISECONDS)
        .onBackpressureDrop()//效果与Drop类型一样
        .subscribeOn(Schedulers.newThread())
        .observeOn(Schedulers.newThread())
        .subscribe(new Consumer<Long>() {
            @Override
            public void accept(Long aLong) throws Exception {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                Log.e(TAG,"onNext : "+(aLong));
            }
        });
	}
}
